package externalSort.Tournament;

import java.awt.color.CMMException;
import java.io.BufferedReader;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class bufferReadManager implements Runnable{

    public int k;

    public List<BufferedReader> bufferedReaders;

    public Lock bufferLock;
    private final Lock queueLock;
    public List<bufferSimulator> buffers;// 整个归并就用这几个buffer了
    public List<BlockingQueue<bufferSimulator>> bufferQueue;

    // val -> way
    private final PriorityQueue<int[]> preReadQueue;

    public final ExecutorService executorService;

    private final Semaphore bufferAvailable;

    public AtomicBoolean allFinished;

    public List<CompletableFuture<Void>> futures;

    public List<BufferedReader> getBufferedReaders() {
        return bufferedReaders;
    }

    public void setBufferedReaders(List<BufferedReader> bufferedReaders) {
        this.bufferedReaders = bufferedReaders;
    }

    public bufferReadManager(int bufferSize, int k){
        this.k=k;
        this.bufferLock=new ReentrantLock();
        this.queueLock=new ReentrantLock();
        this.buffers=new ArrayList<>(k);
        for(int i=0;i<2*k;i++){
            bufferSimulator cur=new bufferSimulator(bufferSize);
            cur.setState(BufferState.IDLE);
            buffers.add(cur);
        }

        bufferQueue = new ArrayList<>(k);
        for (int i=0;i<k;i++){
            bufferQueue.add(new LinkedBlockingQueue<>(bufferSize));
        }
        this.preReadQueue = new PriorityQueue<>(Comparator.comparingInt(a -> a[0]));
        this.executorService= Executors.newFixedThreadPool(k);// 最多k个线程进行读入

        this.bufferAvailable=new Semaphore(2*k);
        this.allFinished=new AtomicBoolean();
        futures=new ArrayList<>();
    }

    public void initRead() throws Exception {

//        System.out.println("===reader初始化===");

        for(int i=0;i<bufferedReaders.size();i++){
            BufferedReader br=bufferedReaders.get(i);
            bufferSimulator curBuffer;

//            System.out.println("[bufferReader] AT initRead : 尝试获取 bufferAvailable");
            bufferAvailable.acquire();
//            System.out.println("[bufferReader] AT initRead : 成功获取 bufferAvailable");
            bufferLock.lock();
            curBuffer = buffers.stream()
                    .filter(buffer -> buffer.getState() == BufferState.IDLE)
                    .findFirst().orElse(null);

            if(curBuffer==null){
                System.out.println("[bufferAllocator]: 没找到闲着的buffer???");
                bufferLock.unlock();
            }
            else{
                curBuffer.setState(BufferState.ACTIVE);
                bufferLock.unlock();
                futures.add(readToBuffer(curBuffer,i,br));
            }
        }
        for(CompletableFuture<Void> future:futures){
            future.get();
        }
        futures.clear();
    }

    @Override
    public void run(){
        try{
            initRead();
            this.allFinished.set(true);// 表示当前的future全部兑现
            // 什么时候结束呢？其实当优先队列为空的时候就可以下班了，因为这说明不可能有任何一个顺串可以再提供数据了
            while(!preReadQueue.isEmpty()){

//                System.out.println("preQueueSize="+preReadQueue.size());

                queueLock.lock();
                int way= Objects.requireNonNull(preReadQueue.poll())[1];
                queueLock.unlock();
//                System.out.println("[bufferReader] AT run : 获取下一个优先队列中的路 "+way);
                bufferAvailable.acquire();
                bufferSimulator curBuffer;
                bufferLock.lock();
                curBuffer=buffers.stream()
                        .filter(buffer->buffer.getState()==BufferState.IDLE)
                        .findFirst().orElse(null);
                if(curBuffer==null){
                    System.out.println("[bufferRead: 没找到闲着的buffer???");
                    bufferLock.unlock();
                }else{
                    curBuffer.setState(BufferState.ACTIVE);
                    bufferLock.unlock();
                    CompletableFuture<Void> future=readToBuffer(curBuffer,way,bufferedReaders.get(way));
                    futures.add(future);
                    future.get();
                }
            }
            this.close();
        }catch (Exception e){
            System.out.println("[bufferAllocator] Exception occurs "+ e.getMessage());
        }
    }

    private record readResult(boolean success, String errorMessage) {}

    private void recordToQueue(bufferSimulator bufferSimulator,int way){
        // 最先被用完的一定是结尾值最小的元素
//        System.out.println("[bufferReader] AT readBuffer : 优先队列存放"+bufferSimulator.back()+" "+way);
        queueLock.lock();
        preReadQueue.add(new int[]{bufferSimulator.back(),way});
        queueLock.unlock();
    }

    private CompletableFuture<Void> readToBuffer(bufferSimulator curBuffer,int way,BufferedReader br) throws Exception {
        return CompletableFuture.supplyAsync(()->{
            try{
                String s=null;
                int cur = 0;
                while(!curBuffer.isFull()){
                    s=br.readLine();
                    cur=(s==null?Integer.MAX_VALUE:Integer.parseInt(s.trim()));
//                    System.out.println("[bufferReader] AT readBuffer : 读取出数字" + cur +",准备写入 way "+way);
                    curBuffer.add(cur);
                    if(cur==Integer.MAX_VALUE){
                        break;
                    }
                }
                // 如果这一个way上已经读到终止符了，那么说明队列就没有必要去维护它了，不添加
                // 这样也应该能保证说当所有文件都读入到终止符之后pq为空
                if(curBuffer.back()!=Integer.MAX_VALUE){
                    recordToQueue(curBuffer,way);
                }
//                System.out.println("[DEBUG] exit block");
                return new readResult(true,null);
            }catch(Exception e){
                return new readResult(false,e.getMessage());
            }
        },executorService).thenRunAsync(()->{
//            System.out.println("[bufferReader] AT readBuffer :  异步回调修改buffer状态为FULL");
            bufferLock.lock();
            curBuffer.setState(BufferState.FULL);
            bufferLock.unlock();
            bufferQueue.get(way).add(curBuffer); // 这个时候再放入队列中
        },executorService);
    }

    public bufferSimulator pollBuffer(int way) throws Exception {
        return bufferQueue.get(way).take();
    }

    public void releaseBuffer(bufferSimulator buffer){
        bufferLock.lock();
        buffer.setState(BufferState.IDLE);
        bufferLock.unlock();
        bufferAvailable.release();
    }

    public void close() throws ExecutionException, InterruptedException {
        for(CompletableFuture<Void> future:futures){
            future.get();
        }
        // 关闭所有 reader
        for (BufferedReader reader : this.bufferedReaders) {
            try {
                reader.close();
            } catch (IOException e) {
                System.out.println("[bufferAllocator] At [close]: Exception occurs"+ e.getMessage());
            }
        }
    }
}